package com.cctc.rds.scheduler.client.schedule.impl;

import com.cctc.rds.nrpc.core.util.JsonUtils;
import com.cctc.rds.scheduler.client.schedule.TaskScheduleClient;
import com.cctc.rds.scheduler.core.raft.TaskRpc;
import com.cctc.rds.scheduler.core.task.Task;
import com.cctc.rds.scheduler.core.task.TaskLog;

import java.util.Arrays;
import java.util.stream.Collectors;

/**
 * 任务调度服务，[客户端] 实现类。
 * 任务 [提交 -> 调度 -> 执行] 流程：
 * Step 1：任务调度器使用方通过任务调度器客户端提供任务类型或任务实例，并将其放入到任务实例工厂内，然后指定任务延迟执行周期，向 Leader 节点提交任务实例。
 * 其中，任务实例会被封装成 [任务日志（penging）] 同步至所有节点，存储到本机 LevelDB；
 * Step 2：Leader 节点扫描本机存储状态为 pending 的任务实例日志放入时间轮（阻塞队列、待调度），更新任务状态为 scheduling；使用时间轮算法，自动将到期
 * 执行的任务实例调度到执行容器内（已调度，待执行）；
 * Step 3：任务调度器客户端自动从 Leader 节点的执行容器中拉取任务并执行，执行完成后向 Leader 节点更新任务执行结果（任务状态：success、fail），并将
 * 更新后的任务实例日志（任务状态：success、fail）同步至所有 Follower 节点；
 * --> 如果任务实例日志状态为 success（执行成功），则结束执行。
 * --> 如果任务实例日志状态为 fail（执行失败），则设置重新调度时间为当前时间，并计算下一次延迟执行时间，更新任务状态为 pending。
 * Step 4：任务调度器使用方通过任务调度器客户端取消任务实例，终止调度执行。
 * 此外，对于 Raft 集群：
 * - 当 Leader 节点挂掉时，对于已经提交到任务调度器服务端的任务，或者没有任务调度器客户端拉取，或者已经被任务调度器客户端拉取但未提交执行结果的任务，都会被
 * 新的 Leader 节点锁定一段时间，然后重置为 pending 状态，重新调度，延迟执行。因此，任务调度器使用方需要自行确保任务实例的幂等执行。
 */
public class DefaultTaskScheduleClient implements TaskScheduleClient {
    private String app;
    private TaskRpc taskRpc;

    public DefaultTaskScheduleClient(String app, TaskRpc taskRpc) {
        this.app = app;
        this.taskRpc = taskRpc;
    }

    private TaskLog newTaskLog(Task<?> task) {
        return TaskLog.builder()
                .taskId(-1)     // 服务端提供
                .taskParam(JsonUtils.toJson(task.getTaskParam()))
                .scheduleTime(System.currentTimeMillis() / 1000)
                .build();
    }

    @Override
    public <T> long submit(Task<T> task, long... periods) {
        TaskLog taskLog = newTaskLog(task);
        taskLog.setExecPriod(Arrays.stream(periods).boxed().collect(Collectors.toList()));
        return taskRpc.save(app, taskLog);
    }

    @Override
    public boolean cancel(long taskId) {
        return taskRpc.cancel(app, taskId);
    }

}
